package com.ming.rabbitmq.four;

import com.ming.rabbitmq.util.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage {

    public static void main(String[] args) throws Exception {

        //ackOne();//发布1000条单独确认消息，耗时： 30240
        //ackBatch();//发布1000条批量确认消息，耗时： 195
        ackAsync();//发布1000条批量确认消息，耗时： 90
    }


    public static void ackOne() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMqUtils.getChannel();

        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            String message = i + "";
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            boolean confirms = channel.waitForConfirms();
           /* if (confirms){

                System.out.println("消息发送完毕: "+message);
            }*/

        }
        long end = System.currentTimeMillis();
        System.out.println("发布1000条单独确认消息，耗时： " + (end - begin));
    }

    public static void ackBatch() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMqUtils.getChannel();

        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            String message = i + "";
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            if (i + 1 % 100 == 0) {
                boolean confirms = channel.waitForConfirms();
            }
            /*if (confirms){
                System.out.println("消息发送完毕: "+message);
            }*/

        }
        long end = System.currentTimeMillis();
        System.out.println("发布1000条异步确认消息，耗时： " + (end - begin));
    }

    public static void ackAsync() throws Exception {
        String QUEUE_NAME = UUID.randomUUID().toString();
        Channel channel = RabbitMqUtils.getChannel();

        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();
        long begin = System.currentTimeMillis();

        ConcurrentSkipListMap<Long, String> skipListMap = new ConcurrentSkipListMap<>();


        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {

            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = skipListMap.headMap(deliveryTag);
                confirmed.clear();
            }else {
                skipListMap.remove(deliveryTag);
            }
            System.out.println("确认的消息： " + deliveryTag);
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = skipListMap.get(deliveryTag);
            System.out.println("未确认的消息： "+message);
            System.out.println("未确认的消息的Tag(消息投递序号)： " + deliveryTag);
        };
        channel.addConfirmListener(ackCallback, nackCallback);
        for (int i = 0; i < 1000; i++) {
            String message = i + "";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            skipListMap.put(channel.getNextPublishSeqNo(), message);
        }

        long end = System.currentTimeMillis();
        System.out.println("发布1000条批量确认消息，耗时： " + (end - begin));
    }


}
